package org.gitos.cloud.server.mq;

import com.alibaba.fastjson.JSON;

import org.gitos.cloud.domain.comm.Constant;
import org.gitos.cloud.domain.device.Device;
import org.gitos.cloud.domain.device.Router;
import org.gitos.cloud.domain.mq.MqMsg;
import org.gitos.cloud.domain.server.Rom;
import org.gitos.cloud.domain.tcp.Bucket;
import org.gitos.cloud.domain.tcp.DeviceId;
import org.gitos.cloud.server.mq.logic.*;
import org.gitos.cloud.server.service.PushToAppService;
import org.gitos.cloud.server.service.RelationService;
import org.gitos.cloud.util.ByteUtil;
import org.gitos.cloud.util.CommonKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Payload;

import java.io.*;
import java.net.URL;
import java.net.URLConnection;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by shizhengchu on 15/12/10. 每个comet接收recieve_exchange的消息,
 * nodeId作为queue name, routekey也是nodeId
 */
@Configuration
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "default", durable = "true"), exchange = @Exchange(value = Constant.SEVER_EXCHANGE, durable = "true"), key = "default"))
public class RecieveManager {

    private static final Logger log = LoggerFactory.getLogger(RecieveManager.class);
    private ExecutorService fixThreadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    @Autowired
    private RelationService relationService;
    @Autowired
    private PushManager pushManager;
    @Autowired
    private PushToAppService pushToAppService;

    @Autowired
    private Router router;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RabbitHandler
    public void onMessage(@Payload String message) {
        log.info(message.toString());
        try {
            MqMsg msg = JSON.parseObject(message.toString(), MqMsg.class);
            DeviceId deviceId = msg.getBucket().getDeviceId();
            Device device = router.fetchDevice(deviceId);
            if (device== null) {
                device=new Device();
                device.setChannelId(msg.getChannelId());
                device.setNodeId(msg.getNodeId());
                device.setDeviceId(deviceId);
            }
            router.registerDevice(device);
            dispatch(msg);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void dispatch(MqMsg msg) {
        switch (msg.getEvent()) {
            //请求鉴权
            case Constant.CMD_UP_AUTH:
                ApplyAuthHandler applyAuthHandler = new ApplyAuthHandler(msg);
                applyAuthHandler.setPushManager(pushManager);
                applyAuthHandler.setChannleId(msg.getChannelId());
                applyAuthHandler.setStringRedisTemplate(stringRedisTemplate);
                applyAuthHandler.setNodeId(msg.getNodeId());
                applyAuthHandler.setStringRedisTemplate(stringRedisTemplate);
                fixThreadPool.execute(applyAuthHandler);
                break;
            //上报认证码
            case Constant.CMD_UP_AUTH_CODE:
                AuthResultHandler authResultHandler = new AuthResultHandler(router);
                authResultHandler.setChannleId(msg.getChannelId());
                authResultHandler.setNodeId(msg.getNodeId());
                authResultHandler.setMsg(msg);
                authResultHandler.setStringRedisTemplate(stringRedisTemplate);
                authResultHandler.setPushManager(pushManager);
                authResultHandler.setStringRedisTemplate(stringRedisTemplate);
                fixThreadPool.execute(authResultHandler);
                break;
            //上报设备信息
            case Constant.CMD_UP_DEVICE_INFO:
                DeviceInfoHandler deviceInfoHandler = new DeviceInfoHandler(msg, router, pushToAppService);
                deviceInfoHandler.setChannleId(msg.getChannelId());
                deviceInfoHandler.setNodeId(msg.getNodeId());
                deviceInfoHandler.setPushManager(pushManager);
                deviceInfoHandler.setRelationService(relationService);
                fixThreadPool.execute(deviceInfoHandler);
                break;
            //设备请求断开连接
            case Constant.CMD_UP_DISCONNECT:
                DisconnectHandler disconnectHandler = new DisconnectHandler(msg, router, pushToAppService);
                disconnectHandler.setChannleId(msg.getChannelId());
                disconnectHandler.setNodeId(msg.getNodeId());
                disconnectHandler.setPushManager(pushManager);
                fixThreadPool.execute(disconnectHandler);
                break;
            //上报心跳
            case Constant.CMD_UP_HEART_BEAT:
                HeartBeatHandler heartBeatHandler = new HeartBeatHandler();
                heartBeatHandler.setPushManager(pushManager);
                heartBeatHandler.setChannleId(msg.getChannelId());
                heartBeatHandler.setNodeId(msg.getNodeId());
                heartBeatHandler.setMsg(msg);
                fixThreadPool.execute(heartBeatHandler);
                break;
            //设备OTA
            case Constant.CMD_UP_OTA:
                DeviceId deviceId1 = msg.getBucket().getDeviceId();
                if(deviceId1==null){
                    break;
                }
                Device device = router.fetchDevice(deviceId1);
                if(device==null||!device.isLogin()){
                    device.setOtaState(Device.OTA_NOT_ALLOWED);
                    router.registerDevice(device);
                    break;
                }
                device.setOtaState(Device.OTA_ALLOWED);
                router.registerDevice(device);
                byte[] content = msg.getBucket().getContent();
                if (content.length != 33 || content[0] != 1) {
                    device.setOtaState(Device.OTA_FAIL);
                    router.registerDevice(device);
                    break;
                }
                String uuid = new String(Arrays.copyOfRange(content, 1, 33));
                Rom rom = relationService.getRomByUuid(uuid);
                if (rom == null){
                    device.setOtaState(Device.OTA_FAIL);
                    router.registerDevice(device);
                    break;
                }
                byte[] bytes = null;
                try {
                    int byteread = 0;
                    int bytesum = 0;
                    URL url = new URL(rom.getUrl());
                    URLConnection conn = url.openConnection();
                    InputStream inStream = conn.getInputStream();

                    int available = inStream.available();

                    byte[] buffer = new byte[available];
                    while ((byteread = inStream.read(buffer)) != -1) {
                        bytesum += byteread;
                    }
                    bytes = buffer;
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                if (bytes== null){
                    device.setOtaState(Device.OTA_FAIL);
                    router.registerDevice(device);
                    break;
                }
                int length = bytes.length;
                int count = 0;
                MqMsg resultMsg = new MqMsg();
                resultMsg.setChannelId(msg.getChannelId());
                resultMsg.setNodeId(msg.getNodeId());
                Bucket bucket = new Bucket();
                bucket.setCmd(Constant.CMD_OTA_TRANSFER);
                DeviceId deviceId = msg.getBucket().getDeviceId();
                bucket.setDeviceId(deviceId);
                //改变OTA状态
                device.setOtaState(Device.OTA_TRANSFER);
                router.registerDevice(device);
                //rom文件过大多次推送，每次推送1024字节
                while (length >= 1024 * count) {
                    byte[] byte_send = Arrays.copyOfRange(bytes, 1024 * count, 1024 * count + 1024);
                    count++;
                    bucket.setContent(byte_send);
                    resultMsg.setBucket(bucket);
                    pushManager.sendMsg(msg.getNodeId(), resultMsg);
                }
                //推送最后不足1024字节的部分
                if (length != 1024 * count) {
                    byte[] byte_send = Arrays.copyOfRange(bytes, 1024 * count - 1024, length);
                    bucket.setContent(byte_send);
                    resultMsg.setBucket(bucket);
                    pushManager.sendMsg(msg.getNodeId(), resultMsg);
                }
                //改变OTA状态
                device.setOtaState(Device.OTA_SUCCESS);
                router.registerDevice(device);
                bucket.setCmd(Constant.CMD_OTA_ROM_SIZE);
                bucket.setContent(ByteUtil.int2Bytes(length));
                resultMsg.setBucket(bucket);
                pushManager.sendMsg(msg.getNodeId(), resultMsg);
                break;
        }
    }
}
